package com.qianbao.finance.sync.modules.elasticsearch.websocket.simulator;

import com.qianbao.finance.sync.modules.elasticsearch.websocket.server.ChatRoom;
import com.qianbao.finance.sync.modules.elasticsearch.websocket.server.ShareBean;
import com.qianbao.finance.sync.modules.elasticsearch.websocket.utils.CodeUtil;
import com.qianbao.finance.sync.modules.elasticsearch.websocket.utils.ZabbixDataUtil;
import org.apache.commons.lang.ArrayUtils;
import org.apache.log4j.Logger;
import org.jboss.netty.buffer.ChannelBuffer;
import org.jboss.netty.buffer.ChannelBuffers;
import org.jboss.netty.channel.Channel;
import org.jboss.netty.channel.ChannelFuture;
import org.jboss.netty.channel.ChannelFutureListener;
import org.jboss.netty.channel.ChannelHandlerContext;
import org.jboss.netty.channel.ExceptionEvent;
import org.jboss.netty.channel.MessageEvent;
import org.jboss.netty.channel.SimpleChannelHandler;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONArray;
import com.alibaba.fastjson.JSONObject;

/**
 * 模拟器接收数据 清洗数据 发送数据
 * 
 * @author WangYuan
 * @date 2016年7月23日下午4:00:45
 */

public class SimulatorHandler extends SimpleChannelHandler {

    private static final String DATA = "agent data";
    private static final String CHECK = "active checks";
    private static final String GROUPNAME = "zoom_group";

    private static final Logger logger = Logger.getLogger(SimulatorHandler.class);

    @Override
    public void exceptionCaught(ChannelHandlerContext ctx, ExceptionEvent e) {
        logger.error(e);
        Channel ch = e.getChannel();
        ch.close();
    }

    @Override
    public void messageReceived(ChannelHandlerContext ctx, MessageEvent e) throws Exception {
        ChannelBuffer buf = (ChannelBuffer) e.getMessage();

        ChatRoom chatRoom;

        byte[] header = new byte[5];
        byte[] len = new byte[8];
        buf.getBytes(0, header, 0, 5);
        buf.getBytes(5, len, 0, 8);
        ArrayUtils.reverse(len);
        long contentLen = CodeUtil.bytes2Long(len);

        byte[] content = new byte[(int) contentLen];
        buf.getBytes(13, content, 0, (int) contentLen);
        buf.clear();
        String jsonContent = new String(content);
        logger.info("header-->" + new String(header));
        logger.info("len-->" + contentLen);
        logger.info("json-->" + jsonContent);

        JSONObject jsonObject = JSON.parseObject(jsonContent);
        String requestType = jsonObject.getString("request");

        /******************* 发送消息开始 ************************/
        /******************* 推送监控指标JSON数据 *******************/
        if (requestType.equals(DATA)) {
            JSONArray jsonArray = jsonObject.getJSONArray("data");
            for (Object json : jsonArray) { // agent是分批发送指标信息的 所以把收到的指标信息分割成json段 发送到Socket client
                /*** logger.info("#" + json); ***/
                JSONObject data = JSON.parseObject(json.toString());
                String host = data.getString("host");
                String chatRoomName = host; // 主机名称为推送标识
                data.put("clock", System.currentTimeMillis()); // Highcharts 不识别Unix时间戳 换成java
                if ((chatRoom = ShareBean.getInstance().getChatRoom().get(chatRoomName)) != null) { // 前台没有Socket链接不发送数据
                    chatRoom.broadcastMessage(GROUPNAME, data.toString());
                }
            }
        }
        /******************* 发送消息结束 ************************/

        String resJsonStr = "";
        if (requestType.equals(CHECK)) {
            String host = jsonObject.getString("host");
            resJsonStr = ZabbixDataUtil.getItems(host, ZabbixDataUtil.getToken());
        } else if (requestType.equals(DATA)) {
            int dataSize = jsonObject.getJSONArray("data").size();
            resJsonStr =
                    "{\"response\":\"success\",\"info\":\"processed:" + dataSize
                            + "; failed: 0; total:" + dataSize + "; seconds spent: 0.003534\"}";
        }
        byte[] respByte = resJsonStr.getBytes();
        long respLen = respByte.length;
        byte[] respLenByte = CodeUtil.long2Bytes(respLen);
        ArrayUtils.reverse(respLenByte);
        logger.info("response str len-->" + resJsonStr.length());
        logger.info("response byte len-->" + respByte.length);

        ChannelBuffer channelBuffer = ChannelBuffers.buffer((int) respLen + 13);

        channelBuffer.writeBytes(header, 0, 5);
        channelBuffer.writeBytes(respLenByte, 0, 8);
        channelBuffer.writeBytes(respByte, 0, (int) respLen);

        Channel ch = e.getChannel();
        ChannelFuture f = ch.write(channelBuffer);

        f.addListener(new ChannelFutureListener() {
            @Override
            public void operationComplete(ChannelFuture future) {
                Channel ch = future.getChannel();
                ch.close();
            }
        });
    }

}
